package com.boot.study.api

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    // 创建批处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行
    env.setParallelism(1)

    // 从外部命令中提取参数
    // windows启动 nc64.exe -L -p 9000
    // --host 127.0.0.1 --port 9000
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = parameterTool.get("host")
    val port: Int = parameterTool.getInt("port")

    // 接受socket文本流
    val dataStream: DataStream[String] = env.socketTextStream(host, port)
    // 转换处理分组
    val resultDataStream: DataStream[SensorReading] = dataStream.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    resultDataStream
      .keyBy(_.id)
      .process(new TempIncreWarning(10000L))
      .print("waring test")

    // 启动任务执行
    env.execute("procrss function test")
  }
}

// 实现自定义的KeyedProcessFunction
class TempIncreWarning(interval: Long) extends KeyedProcessFunction[String, SensorReading, String] {
  // 定义状态 保存上一个温度值进行比较，保存定时器的时间搓用于删除
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  lazy val timeTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timeTs", classOf[Long]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
    // 先取出状态
    val lastTemp: Double = lastTempState.value()
    val timeTs: Long = timeTsState.value()

    // 更新状态值
    lastTempState.update(value.temperature)

    // 当前温度值和上次进行比较
    if (value.temperature > lastTemp && timeTs == 0) {
      // 如果温度上升且没有定时器，那么定义当前时间搓10s之后的定时器
      val ts: Long = ctx.timerService().currentProcessingTime() + interval
      // 定义定时器
      ctx.timerService().registerProcessingTimeTimer(ts)
      // 状态保存
      timeTsState.update(ts)
    } else if (value.temperature < lastTemp) {
      // 如果温度下降删除定时器
      ctx.timerService().deleteProcessingTimeTimer(timeTs)
      // 清空
      timeTsState.clear()
    }
  }

  // 触发定时器
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect("传感器" + ctx.getCurrentKey + "的温度" + (interval / 1000 + "秒连续上升"))
    timeTsState.clear()
  }
}

// KeyedProcessFunction功能测试
class MyKeyedProcessFunction extends KeyedProcessFunction[String, SensorReading, String] {

  var myState: ValueState[Int] = _

  override def open(parameters: Configuration): Unit = {
    myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("myState", classOf[Int]))
  }

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
    ctx.getCurrentKey // 当前分组的key
    ctx.timestamp() // 时间搓
    ctx.timerService().currentWatermark() // 水位线
    ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 60000L) // 注册定时器
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {

  }
}

